/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Arrays;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.reliable.events.RdAckInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdAckResponseServiceInvoiceEvent;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.RdRecoveryResponse;
import cgl.narada.service.reliable.events.RdRetransmissionServiceEvent;

/**
   This class is responsible for ensuring retransmissions of NAKs and recovery
   requests. This thread needs to have reconfigurable parameters which ensures
   that parameters pertaining to retransmission can be easily specified.

    @author Shrideep Pallickara
    $Date$
    $Revision$
*/

public class InvoicerThread extends Thread implements RdDebugFlags {
  
  /** requestId -> recoveryRequest */
  private Hashtable pendingRecoveryRequests;

  /** requestId -> nakRequest */
  private Hashtable pendingNakRequests;

  /** templateId --> sync */
  private Hashtable recoveringOnTemplate;
  
  private Object syncObject;
  
  private long REISSUE_INTERVAL = 4000;
  
  private long THREAD_SLEEP_INTERVAL = 300;

  private int entityId;
  private EventProducer producer;
  private String rdsString = "ReliableDeliveryService/";
  
  private NakProcessor nakProcessor;
  private boolean keepLooping = true;
  private boolean interrupt = false;
  private String moduleName = "InvoicerThread: ";
  

  public InvoicerThread(int entityId, EventProducer producer) {
    pendingRecoveryRequests = new Hashtable();
    pendingNakRequests = new Hashtable();
    recoveringOnTemplate = new Hashtable();
    syncObject = new Object();
    this.entityId = entityId;
    this.producer = producer;
    nakProcessor = new NakProcessor(entityId, producer);
    setName("InvoicerThread");
  }

  public void processMissedSequences(int templateId, long[] missedSequences) {
    nakProcessor.processMissedSequences(templateId, missedSequences);

    synchronized(syncObject) {
      syncObject.notifyAll();
    };
  }

  public void processMissedSequence(int templateId, long missedSequence) {
    nakProcessor.processMissedSequence(templateId, missedSequence);

    synchronized(syncObject) {
      syncObject.notifyAll();
    };
  }


  public void
  processRetransmittedEvent(RdRetransmissionServiceEvent retransmitEvent) {
    nakProcessor.processRetransmittedEvent(retransmitEvent);
  }

  public void issueRecoveryRequest(RdRecoveryEntityRequest recoveryRequest) 
    throws ServiceException {
    if (InvoicerThread_Debug) {
      System.out.println(moduleName + "Issuing recovery request ");
    }
    int templateId = recoveryRequest.getTemplateId();
    NBEvent nbEvent = 
      producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			     rdsString + templateId + "/Recovery", 
			     recoveryRequest.getBytes());
    
    NBEventGenerator.setEventType(nbEvent, 
				  NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
    producer.publishEvent(nbEvent);

  }


  public void manageRecoveryRequest(RdRecoveryEntityRequest recoveryRequest)
    throws ServiceException {
    int templateId = recoveryRequest.getTemplateId();
    Object templateKey = new Integer(templateId);
    long recoveryId = recoveryRequest.getRecoveryRequestId();
    Object recoveryKey = new Long(recoveryId);

    if (recoveringOnTemplate.containsKey(templateKey)) {
      if (pendingRecoveryRequests.containsKey(recoveryKey) ) {
	System.out.println(moduleName + "Trying to reissue a pending recovery"
			   + " request " + recoveryRequest);
      } else {
	System.out.println(moduleName + "Trying to issue a recovery request" +
			   " while one has been initiated for template (" +
			   templateId + ")");
      }
      return;
    }
    
    /** Initialize the recovering on template */
    recoveringOnTemplate.put(templateKey, new Long(-1));
    
    pendingRecoveryRequests.put(recoveryKey, recoveryRequest);
    
    synchronized(syncObject) {
      syncObject.notifyAll();
    };
  }
  
  
  
  public void manageRecoveryResponse(RdRecoveryResponse recoveryResponse) 
    throws ServiceException {
    Object recoveryKey = new Long(recoveryResponse.getRecoveryRequestId());
    int templateId = recoveryResponse.getTemplateId(); 
    Object templateKey = new Integer(templateId);
    
    if (pendingRecoveryRequests.containsKey(recoveryKey)) {
      RdAckResponseServiceInvoiceEvent ackResponse = 
	recoveryResponse.getSubscriberRecoveryInfo().getAckResponse();
      long donotAckUntilSyncHasReached;
      
      
      if (ackResponse.containsMissedSequenceInfo()) {
	if (ackResponse.containsSingleSequenceInfo() ) {
	  donotAckUntilSyncHasReached = ackResponse.getMissedSequence();
	} else {
	  long[] missedSequences = ackResponse.getMissedSequences();
	  Arrays.sort(missedSequences);
	  int largestIndex = missedSequences.length -1;
	  donotAckUntilSyncHasReached = missedSequences[largestIndex];
	}
	Object donotAckUntilKey = new Long(donotAckUntilSyncHasReached);
	recoveringOnTemplate.put(templateKey, donotAckUntilKey);
	System.out.println(moduleName + "Will ensure that ACKS to template (" +
			   templateId + ") are attentuated if the sequence is"
			   + " GREATER than " + donotAckUntilSyncHasReached);
      } else {
	/** Since there were no missed sequences recovery is complete */
	System.out.println(moduleName + "There were no missing sequences on " +
			   "template [" + templateId + "]. Recovery process is"
			   + " thus complete!");
	recoveringOnTemplate.remove(templateKey);
      }
      
      
      
      pendingRecoveryRequests.remove(recoveryKey);
      return;
    }
    
    System.out.println(moduleName + "Unknown recovery (" + recoveryKey + 
		       ") identifier. ");
  }
  
  
  

  
  public void manageNakRequest() {
    
  }
  
  /** Check to see if the sync advancement implies that recovery on the
      recovering template is complete. If the advanceSync is equal to
      the associated donotAckUntil, recovery is considered to be complete.*/
  public void processSyncAdvance(int templateId, long advanceSync) {
    Object templateKey = new Integer(templateId);
    if (recoveringOnTemplate.containsKey(templateKey)) {
      long donotAckUntil = 
	((Long)recoveringOnTemplate.get(templateKey)).longValue();
      if (donotAckUntil == advanceSync) {
	recoveringOnTemplate.remove(templateKey);
	System.out.println(moduleName + "Recovery on template (" +
			   templateId + ") is complete!");
      }
    }

    nakProcessor.processSyncAdvance(templateId, advanceSync);
  }
  
  /** This method creates the appropriate AckInvoiceEntityEvent. If this
      method returns a NULL it implies that there is nothing that needs to
      be acknowledged at the present time. One may have received events, but
      until recovery is complete we don't ack events received while 
      "recovering" --- this ensures that there is not a retransmission 
      explosion. 
      
      Lets say a client leaves when it's sync is 400.
      After a few hours when it reconnects it receives event with seq -> 800
      Now also, let us say that recovery response indicates that we have 
      missed 250 events 500-750
      When we ACK retransitted events we will be ACKing 800 also since it is 
      not removed from the ACK set till the syncpoint for a template has 
      actually been advanced to 800.

      So we keep getting ACKResponses which will have missed events between
      whatever was ACKed (say 510) and 800. This will trigger a NAK and so 
      forth. The process then repeats itself over and over. 
      
      This method addresses this problem and fixes it.
*/
  public RdAckInvoiceEntityEvent 
  createAckInvoiceEvent(int templateId, 
			Vector pendingAcksForTemplate) {
    /** We have to make sure that we do not acknowledge certain ack's. 
	Specifically we check to see if "templateId" is contained in recovering
	 on template. 
	 (a) If the value is -1, we don't issue an ACK at all, since  we are 
	 recovering on the template in question and we haven't received a 
	 recovery response yet.
	 (b) If the value is something other than -1, which would mean that 
	 donotAckUntilSyncHasReached has been initialized by a recoveryResponse
	 We need to make sure that sequences > donotAckUntilSyncHasReached in 
	 pendingAcks should not be ACKed.  
    */
    Object templateKey = new Integer(templateId);
    long donotAckUntil = 0;
    if (recoveringOnTemplate.containsKey(templateKey)) {
      donotAckUntil = 
	((Long)recoveringOnTemplate.get(templateKey)).longValue(); 
      if (InvoicerThread_Debug) {
	System.out.println(moduleName + "Will not ack on RECOVERING template ("
			   + templateKey + ") if the value you are trying to" +
			   " ACK is >= " + donotAckUntil);
      }
    }
    
    if (donotAckUntil < 0) {
      System.out.println(moduleName + "Recovering on template (" + templateId +
			 "), a request has been previously issued. However, " +
			 "we have NOT yet received a recovery response. " + 
			 "So NO ACKS should/will be issued on this template");
      return null;
    }
    
    
    int numOfAcksToIssue = 0;
    long[] listOfAllAcks = new long[pendingAcksForTemplate.size()];
    int totalNumOfAcks  =  listOfAllAcks.length;
    
    for (int i=0; i < totalNumOfAcks; i++) {
      listOfAllAcks[i] = 
	((Long)pendingAcksForTemplate.elementAt(i)).longValue();

      if (listOfAllAcks[i] <= donotAckUntil) {
	numOfAcksToIssue++;
      }
    }

    long seqNum = 0;
    long[] seqNums = null;
    RdAckInvoiceEntityEvent ackInvoice = null;

    /** This part deals with the case where we are not actually recovering
	on the template */
    if (donotAckUntil == 0) {
      if (totalNumOfAcks == 1) {
	seqNum = listOfAllAcks[0];
	ackInvoice = new RdAckInvoiceEntityEvent(templateId, entityId, seqNum);
      } else {
	seqNums = listOfAllAcks;
	ackInvoice = 
	  new RdAckInvoiceEntityEvent(templateId, entityId, seqNums);
      } /*(totalNumOfAcks == 1 )*/
      return ackInvoice;
    } /* if (donotAckUntil == 0) */
    
    
    
    /** This part deals with case where we are actually recovering on the
	template in question */
    if (numOfAcksToIssue == 0) {
      return null;
    }

    seqNums = new long[numOfAcksToIssue];
    int index=0;
    for (int i=0; i < totalNumOfAcks; i++) {
      if (listOfAllAcks[i] <= donotAckUntil) {
	seqNums[index] = listOfAllAcks[i];
	index++;
	if (index == numOfAcksToIssue) {
	  break;
	}
      }
    }
    
    

    if (numOfAcksToIssue == 1) {
      seqNum = seqNums[0];
      ackInvoice = new RdAckInvoiceEntityEvent(templateId, entityId, seqNum);
    } else {
      ackInvoice = 
	new RdAckInvoiceEntityEvent(templateId, entityId, seqNums);
    } /*(totalNumOfAcks == 1 )*/

    return ackInvoice;
  }
  

  private void checkAndReissuePendingRequests() {
    if (pendingRecoveryRequests.size() == 0) {
      return;
    }
    
    Enumeration keys = pendingRecoveryRequests.keys();

    while (keys.hasMoreElements()) {
      Long timeKey = (Long)keys.nextElement();
      long timeIssued = timeKey.longValue();
      long elapsedTime = System.currentTimeMillis() - timeIssued;
      
      if (elapsedTime > REISSUE_INTERVAL) {
	/** Proceed to reissue the recovery request */
	RdRecoveryEntityRequest recoveryRequest = 
	  (RdRecoveryEntityRequest) pendingRecoveryRequests.get(timeKey);
	try {
	  issueRecoveryRequest(recoveryRequest);
	  pendingRecoveryRequests.remove(timeKey);
	  timeKey = new Long(System.currentTimeMillis());
	  pendingRecoveryRequests.put(timeKey, recoveryRequest);
	} catch (ServiceException serEx) {
	  System.out.println(moduleName + serEx);
	}
      }
      
      
    }
  }

  private void checkAndReissueRecoveryRequests() {
    


  }
  
  
  private void checkAndReissueNakRequests() {
    if (!nakProcessor.hasPendingNaksToIssue()) {
      return;
    }
    nakProcessor.proceedToIssuePendingNaks();
  }
  
  
  public void run() {
    while (true) {
      /** If the size of the recovery requests or nak requests is zero, 
	  wait */
      try {
	if (pendingRecoveryRequests.size() == 0 && 
	    !nakProcessor.hasPendingNaksToIssue()) {
	  synchronized(syncObject) {
	    syncObject.wait();
	  };
	} 
	
	
	/** If the size of the recovery/nak requests is NOT zero, wakeup at 
	    regular intervals to check to see if there are any requests that 
	    need to be reissued. */
	checkAndReissuePendingRequests();
	checkAndReissueNakRequests();
	sleep(THREAD_SLEEP_INTERVAL);
      } catch (InterruptedException ie) {
	if (!interrupt) {
	  ie.printStackTrace();
	}
      }/*end try-catch */ 
      
    } /* end while(true) */
    
  }/* end run() */
  
  
  public void terminateService() throws ServiceException {
    keepLooping = false;
    
    synchronized(syncObject) {
      syncObject.notify();
    };
    /** If the thread is waiting, make sure that once it wakesup it shutsdown
	properly */
    THREAD_SLEEP_INTERVAL=10;
    /** If the thread is currently sleeping, we need to force it to shutdown 
	by first interrupting the thread */
    interrupt = true;
    interrupt();
  }

}
